package cn.tjjcy.service;

import cn.hutool.core.date.DateUtil;
import cn.tjjcy.common.DmMsg;
import cn.tjjcy.common.GzResult;
import cn.tjjcy.dao.RealGzConfMapper;
import cn.tjjcy.domain.RealGzConfEntity;
import cn.tjjcy.service.process.AbstractProcessService;
import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import java.util.Date;
import java.util.List;
import java.util.Objects;

@Component
@Slf4j
public class YwscConsumer {


    @Autowired
    private ProcessorSelector processorSelector;

    @Autowired
    private RealGzConfMapper realGzConfMapper;

    @Autowired
    private YwscGzLogService ywscGzLogService;

    @KafkaListener(topics = {"${spring.kafka.consumer.topic}"}, groupId = "${spring.kafka.consumer.group-id}", concurrency = "1")
    public void onEvent(String msg) {
        try {
        DmMsg value = JSON.parseObject(msg, DmMsg.class);
        String table = value.getTable();
        String opType = value.getOp_type();
        if (StringUtils.isEmpty(table) || Objects.isNull(value.getAfter())) {
            return;
        }
        String sourceTableName = table.split("\\.")[1];
        //查询规则
        QueryWrapper<RealGzConfEntity> queryWrapper = new QueryWrapper();
        queryWrapper.eq("table_name", sourceTableName);
        queryWrapper.like("op_type", opType);
        queryWrapper.orderByAsc("CAST(id as integer)");
        List<RealGzConfEntity> realGzConfEntities = realGzConfMapper.selectList(queryWrapper);
        if (CollectionUtils.isEmpty(realGzConfEntities)) {
            return;
        }
        log.info("表[{}] 操作:[{}] 已配置 [{}] 个规则 开始处理 msg:[{}]", sourceTableName,opType,realGzConfEntities.size(),msg);
            Thread.sleep(3000);
            realGzConfEntities.stream().forEach((entity) -> {
                AbstractProcessService abstractProcessService = processorSelector.selectProcess(entity);
                //通用判断
                GzResult gzResult = abstractProcessService.gzJudge(value, entity);
                if(Objects.isNull(gzResult)){
                    return;
                }


                //记录日志
                String logId = DateUtil.format(new Date(), "yyyyMMddHHmmssSSS");
                ywscGzLogService.initLog(logId,gzResult, value,entity);
                if(!gzResult.getGzResult()){
                    return;
                }



                //规则执行结果
                abstractProcessService.gzResult(gzResult, value, entity);


                //记录日志
                ywscGzLogService.update(logId,gzResult);

            });
            //顺序执行规则
        } catch (InterruptedException e) {
            log.error("YwscConsumer error",e);
        }
    }
}
